#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdbool.h>
#include <unistd.h>
#include <errno.h>
#include "MQTTClient.h"
#include "shmem.h"
#include "cJSON.h"
#include "msg_queue_peer.h"

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC_UP "/app/data/up"
#define TOPIC_DOWN "/app/data/down"
#define QOS 1
#define TIMEOUT 10000L

#define STM32 3  // STM32的消息队列的消息类型
#define MODBUS 1 // MODBUS的消息队列的消息类型

// 共享内存相关变量
struct shm_param para;
int *p;
struct node_data *n_data;

cJSON *Pointroot;
cJSON *report;

int flag_type0 = 0; // 刷新上报标志位
int size;

// 确保结构体1字节对齐
union val_t
{
    bool bval;
    int ival;
    float fval;
};

struct node_data
{
    int key;
    int type;
    int dev_type;
    union val_t old_val;
    union val_t new_val;
    int ret;
};

struct setdata
{
    long devtype;
    int key;
    int val_type; // 添加值类型标识
    union val_t val;
};

volatile MQTTClient_deliveryToken deliveredtoken;

// ================== 函数声明 ==================
void delivered(void *context, MQTTClient_deliveryToken dt);
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void connlost(void *context, char *cause);
void publish_data(MQTTClient client, cJSON *data, bool is_full_report);
void update_shared_memory(int key, int val_type, union val_t new_val);
void reload_configuration(cJSON *newreport);
void initialize_shared_memory();
void parse_point_table(int fd);
int initialize_mqtt(MQTTClient *client);
void handle_reporting_mode(MQTTClient client, int pubtype, int period);

// ================== 工具函数 ==================
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

void publish_data(MQTTClient client, cJSON *data, bool is_full_report)
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddNumberToObject(root, "type", 1);
    cJSON_AddNumberToObject(root, "result", 0);
    cJSON_AddItemToObject(root, "data", data);

    char *payload = cJSON_Print(root);
    if (payload)
    {
        MQTTClient_message pubmsg = MQTTClient_message_initializer;
        pubmsg.payload = payload;
        pubmsg.payloadlen = (int)strlen(payload);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;

        MQTTClient_deliveryToken token;
        MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);

        if (is_full_report)
        {
            printf("Published full report: %s\n", payload);
        }
        else
        {
            printf("Published delta report: %s\n", payload);
        }

        free(payload);
    }
    cJSON_Delete(root);
}

void update_shared_memory(int key, int val_type, union val_t new_val)
{
    for (int i = 0; i < *p; i++)
    {
        if (n_data[i].key == key)
        {
            switch (val_type)
            {
            case 1: // BOOL
                n_data[i].old_val.bval = n_data[i].new_val.bval;
                n_data[i].new_val.bval = new_val.bval;
                printf("更新值: key=%d, old=%d, new=%d\n",
                       key, n_data[i].old_val.bval, n_data[i].new_val.bval);
                break;
            case 2: // INT
                n_data[i].old_val.ival = n_data[i].new_val.ival;
                n_data[i].new_val.ival = new_val.ival;
                printf("更新值: key=%d, old=%d, new=%d\n",
                       key, n_data[i].old_val.ival, n_data[i].new_val.ival);
                break;
            case 3: // FLOAT
                n_data[i].old_val.fval = n_data[i].new_val.fval;
                n_data[i].new_val.fval = new_val.fval;
                printf("更新值: key=%d, old=%.2f, new=%.2f\n",
                       key, n_data[i].old_val.fval, n_data[i].new_val.fval);
                break;
            }
            return;
        }
    }
    printf("Warning: Key %d not found in shared memory\n", key);
}

void reload_configuration(cJSON *newreport)
{
    cJSON_ReplaceItemInObject(Pointroot, "report", newreport);

    int fd = open("/mnt/config/node.json", O_WRONLY | O_TRUNC);
    if (fd < 0)
    {
        perror("open fault");
        return;
    }

    char *config_str = cJSON_Print(Pointroot);
    if (config_str)
    {
        ssize_t written = write(fd, config_str, strlen(config_str));
        if (written < 0)
        {
            perror("write failed");
        }
        else
        {
            printf("成功修改上报模式\n");
        }
        free(config_str);
    }

    close(fd);
    cJSON_Delete(newreport);
}

// ================== 主功能函数 ==================
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Received message on topic %s: %.*s\n",
           topicName, message->payloadlen, (char *)message->payload);

    cJSON *root = cJSON_Parse(message->payload);
    if (!root)
    {
        fprintf(stderr, "JSON parse error\n");
        MQTTClient_freeMessage(&message);
        MQTTClient_free(topicName);
        return 1;
    }

    cJSON *type_obj = cJSON_GetObjectItem(root, "type");
    if (!type_obj)
    {
        fprintf(stderr, "Missing 'type' in message\n");
        cJSON_Delete(root);
        MQTTClient_freeMessage(&message);
        MQTTClient_free(topicName);
        return 1;
    }

    int msg_type = type_obj->valueint;
    int ret_val = 1;

    switch (msg_type)
    {
    case 1: // 刷新上报请求
        flag_type0 = 1;
        printf("请求刷新数据\n");
        break;

    case 2:
    { // 修改设备值
        cJSON *data = cJSON_GetObjectItem(root, "data");
        if (!data)
        {
            fprintf(stderr, "Missing 'data' in message\n");
            break;
        }

        cJSON *key_obj = cJSON_GetObjectItem(data, "key");
        cJSON *val_obj = cJSON_GetObjectItem(data, "val");
        if (!key_obj || !val_obj)
        {
            fprintf(stderr, "Missing key or value in data\n");
            break;
        }

        int key = key_obj->valueint;
        printf("Key: %d\n", key);

        struct setdata d = {0};
        int key_found = 0;

        for (int i = 0; i < *p; i++)
        {
            if (n_data[i].key == key)
            {
                key_found = 1;
                d.devtype = n_data[i].dev_type;
                d.key = key;
                d.val_type = n_data[i].type; // 设置值类型标识

                switch (n_data[i].type)
                {
                case 1:
                    d.val.bval = cJSON_IsTrue(val_obj) ? true : false;
                    printf("msg_queue_send: key=%d, val=%d\n", key, d.val.bval);
                    break;
                case 2:
                    d.val.ival = val_obj->valueint;
                    printf("msg_queue_send: key=%d, val=%d\n", key, d.val.ival);
                    break;
                case 3:
                    d.val.fval = (float)val_obj->valuedouble;
                    printf("msg_queue_send: key=%d, val=%.2f\n", key, d.val.fval);
                    break;
                }
                break;
            }
        }

        if (!key_found)
        {
            printf("No such device!! Key: %d\n", key);
            break;
        }

        // 发送设置命令
        if (msg_queue_send("msg", &d, sizeof(struct setdata), 0) < 0)
        {
            perror("msg_queue_send failed");
            break;
        }

        // 接收响应
        struct setdata b = {0};
        long msg_type = (d.devtype == STM32) ? STM32 : MODBUS;

        if (msg_queue_recv("msg", &b, sizeof(struct setdata), msg_type, 0) < 0)
        {
            perror("msg_queue_recv failed");
            break;
        }

        printf("msgrcv : devtype=%ld, key=%d, val_type=%d\n",
               b.devtype, b.key, b.val_type);

        // 更新共享内存
        update_shared_memory(b.key, b.val_type, b.val);
        break;
    }

    case 3:
    {   // 修改上报模式
        cJSON *data = cJSON_GetObjectItem(root, "data");
        if (!data)
        {
            fprintf(stderr, "Missing 'data' in message\n");
            break;
        }

        cJSON *change_type = cJSON_GetObjectItem(data, "type");
        if (!change_type)
        {
            fprintf(stderr, "Missing 'type' in data\n");
            break;
        }

        cJSON *newreport = cJSON_CreateObject();
        cJSON_AddItemToObject(newreport, "type", cJSON_CreateNumber(change_type->valueint));

        if (change_type->valueint == 2)
        { // 周期上报
             printf("修改模式为：周期上报\n");
            cJSON *change_period = cJSON_GetObjectItem(data, "period");
            if (change_period)
            {
                printf("上报周期: %d\n", change_period->valueint);
                cJSON_AddItemToObject(newreport, "period", cJSON_CreateNumber(change_period->valueint));
            }
            else
            {
                fprintf(stderr, "Missing period for periodic reporting\n");
            }
        }
        else
        { // 其他上报模式
            cJSON *tmp = cJSON_GetObjectItem(report, "period");
            if (tmp)
            {
                cJSON_AddItemToObject(newreport, "period", cJSON_CreateNumber(tmp->valueint));
            }

            if (change_type->valueint == 0)
            {
                printf("修改模式为：刷新上报\n");
            }
            else
            {
                printf("修改模式为：变化上报\n");
            }
        }

        reload_configuration(newreport);
        break;
    }

    default:
        fprintf(stderr, "Unknown message type: %d\n", msg_type);
        break;
    }

    cJSON_Delete(root);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return ret_val;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("Cause: %s\n", cause);
}

void initialize_shared_memory()
{
    // 初始化共享内存
    shm_init(&para, "home", 1024);
    p = shm_getaddr(&para);
    n_data = (struct node_data *)(p + 1);
    printf("Shared memory initialized\n");
}

void parse_point_table(int fd)
{
    char buf[2048];
    ssize_t bytes_read = read(fd, buf, sizeof(buf) - 1);
    if (bytes_read <= 0)
    {
        perror("Failed to read point table");
        return;
    }
    buf[bytes_read] = '\0';

    Pointroot = cJSON_Parse(buf);
    if (!Pointroot)
    {
        fprintf(stderr, "Failed to parse point table JSON\n");
        return;
    }

    report = cJSON_GetObjectItem(Pointroot, "report");
    if (!report)
    {
        fprintf(stderr, "Missing 'report' in point table\n");
        return;
    }

    cJSON *version = cJSON_GetObjectItem(Pointroot, "version");
    if (version)
    {
        printf("Version: %s\n", version->valuestring);
    }

    // 计算数据点数量
    int num_data = 0;
    cJSON *stm32 = cJSON_GetObjectItem(Pointroot, "stm32");
    if (stm32)
    {
        cJSON *stm32_data = cJSON_GetObjectItem(stm32, "data");
        if (stm32_data)
            num_data += cJSON_GetArraySize(stm32_data);
    }

    cJSON *modbus = cJSON_GetObjectItem(Pointroot, "modbus");
    if (modbus)
    {
        cJSON *modbus_data = cJSON_GetObjectItem(modbus, "data");
        if (modbus_data)
            num_data += cJSON_GetArraySize(modbus_data);
    }

    *p = num_data;
    printf("Point table parsed, %d data points\n", num_data);

    // 填充共享内存数据结构
    int index = 0;
    if (stm32)
    {
        cJSON *stm32_data = cJSON_GetObjectItem(stm32, "data");
        if (stm32_data)
        {
            for (int i = 0; i < cJSON_GetArraySize(stm32_data); i++)
            {
                cJSON *item = cJSON_GetArrayItem(stm32_data, i);
                cJSON *key = cJSON_GetObjectItem(item, "key");
                cJSON *type = cJSON_GetObjectItem(item, "type");
                if (key && type)
                {
                    n_data[index].key = key->valueint;
                    n_data[index].type = type->valueint;
                    n_data[index].dev_type = STM32;
                    index++;
                }
            }
        }
    }

    if (modbus)
    {
        cJSON *modbus_data = cJSON_GetObjectItem(modbus, "data");
        if (modbus_data)
        {
            for (int i = 0; i < cJSON_GetArraySize(modbus_data); i++)
            {
                cJSON *item = cJSON_GetArrayItem(modbus_data, i);
                cJSON *key = cJSON_GetObjectItem(item, "key");
                cJSON *type = cJSON_GetObjectItem(item, "type");
                if (key && type)
                {
                    n_data[index].key = key->valueint;
                    n_data[index].type = type->valueint;
                    n_data[index].dev_type = MODBUS;
                    index++;
                }
            }
        }
    }
}

int initialize_mqtt(MQTTClient *client)
{
    MQTTClient_create(client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(*client, NULL, connlost, msgarrvd, delivered);

    int rc = MQTTClient_connect(*client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS)
    {
        fprintf(stderr, "Failed to connect, return code %d\n", rc);
        return -1;
    }

    MQTTClient_subscribe(*client, TOPIC_DOWN, QOS);
    printf("MQTT connected and subscribed\n");
    return 0;
}

void handle_reporting_mode(MQTTClient client, int pubtype, int period)
{
    static int time_counter = 0;
    static bool initial_report_sent = false;
    int changes_detected = 0;

    cJSON *data = cJSON_CreateArray();
    if (!data)
        return;

    switch (pubtype)
    {
    case 0: // 刷新上报
        if (flag_type0)
        {
            for (int i = 0; i < *p; i++)
            {
                cJSON *item = cJSON_CreateObject();
                if (!item)
                    continue;

                cJSON_AddNumberToObject(item, "key", n_data[i].key);

                char val_str[32];
                switch (n_data[i].type)
                {
                case 1:
                    snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.bval);
                    break;
                case 2:
                    snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.ival);
                    break;
                case 3:
                    snprintf(val_str, sizeof(val_str), "%.2f", n_data[i].new_val.fval);
                    break;
                default:
                    snprintf(val_str, sizeof(val_str), "unknown");
                }
                cJSON_AddStringToObject(item, "val", val_str);
                cJSON_AddItemToArray(data, item);
            }
            publish_data(client, data, true);
            flag_type0 = 0;
        }
        else
        {
            cJSON_Delete(data);
        }
        break;

    case 1: // 变化上报
        if (!initial_report_sent)
        {
            // 首次发送全量数据
            for (int i = 0; i < *p; i++)
            {
                cJSON *item = cJSON_CreateObject();
                if (!item)
                    continue;

                cJSON_AddNumberToObject(item, "key", n_data[i].key);

                char val_str[32];
                switch (n_data[i].type)
                {
                case 1:
                    snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.bval);
                    break;
                case 2:
                    snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.ival);
                    break;
                case 3:
                    snprintf(val_str, sizeof(val_str), "%.2f", n_data[i].new_val.fval);
                    break;
                }
                cJSON_AddStringToObject(item, "val", val_str);
                cJSON_AddItemToArray(data, item);
            }
            publish_data(client, data, true);
            initial_report_sent = true;
        }
        else
        {
            // 检测变化
            for (int i = 0; i < *p; i++)
            {
                bool changed = false;
                switch (n_data[i].type)
                {
                case 1:
                    changed = (n_data[i].old_val.bval != n_data[i].new_val.bval);
                    if (changed)
                        n_data[i].old_val.bval = n_data[i].new_val.bval;
                    break;
                case 2:
                    changed = (n_data[i].old_val.ival != n_data[i].new_val.ival);
                    if (changed)
                        n_data[i].old_val.ival = n_data[i].new_val.ival;
                    break;
                case 3:
                    changed = (n_data[i].old_val.fval != n_data[i].new_val.fval);
                    if (changed)
                        n_data[i].old_val.fval = n_data[i].new_val.fval;
                    break;
                }

                if (changed)
                {
                    cJSON *item = cJSON_CreateObject();
                    if (!item)
                        continue;

                    cJSON_AddNumberToObject(item, "key", n_data[i].key);

                    char val_str[32];
                    switch (n_data[i].type)
                    {
                    case 1:
                        snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.bval);
                        break;
                    case 2:
                        snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.ival);
                        break;
                    case 3:
                        snprintf(val_str, sizeof(val_str), "%.2f", n_data[i].new_val.fval);
                        break;
                    }
                    cJSON_AddStringToObject(item, "val", val_str);
                    cJSON_AddItemToArray(data, item);
                    changes_detected++;
                }
            }

            if (changes_detected > 0)
            {
                publish_data(client, data, false);
            }
            else
            {
                cJSON_Delete(data);
                static int counter = 0;
                if (++counter % 10 == 0)
                {
                    printf("No data changes detected\n");
                }
            }
        }
        sleep(1);
        break;

    case 2: // 周期上报
        sleep(1);
        time_counter++;

        if (time_counter >= period)
        {
            time_counter = 0;
            for (int i = 0; i < *p; i++)
            {
                cJSON *item = cJSON_CreateObject();
                if (!item)
                    continue;

                cJSON_AddNumberToObject(item, "key", n_data[i].key);

                char val_str[32];
                switch (n_data[i].type)
                {
                case 1:
                    snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.bval);
                    break;
                case 2:
                    snprintf(val_str, sizeof(val_str), "%d", n_data[i].new_val.ival);
                    break;
                case 3:
                    snprintf(val_str, sizeof(val_str), "%.2f", n_data[i].new_val.fval);
                    break;
                }
                cJSON_AddStringToObject(item, "val", val_str);
                cJSON_AddItemToArray(data, item);
            }
            publish_data(client, data, true);
        }
        else
        {
            cJSON_Delete(data);
            printf("Waiting for report... (%d/%d)\n", time_counter, period);
        }
        break;
    }
}

// ================== 主函数 ==================
int main(int argc, char *argv[])
{
    MQTTClient client;
    int fd;
    int pubtype = 0;
    int period = 0;

    // 初始化共享内存
    initialize_shared_memory();

    // 解析点表
    fd = open("/mnt/config/node.json", O_RDONLY);
    if (fd < 0)
    {
        perror("Failed to open point table");
        return EXIT_FAILURE;
    }

    parse_point_table(fd);
    close(fd);

    // 获取上报配置
    if (report)
    {
        cJSON *type_obj = cJSON_GetObjectItem(report, "type");
        cJSON *period_obj = cJSON_GetObjectItem(report, "period");

        if (type_obj)
            pubtype = type_obj->valueint;
        if (period_obj)
            period = period_obj->valueint;
    }

    printf("Reporting mode: %s, Period: %d\n",
           (pubtype == 0) ? "Refresh" : (pubtype == 1) ? "Change-based"
                                                       : "Periodic",
           period);

    // 初始化MQTT
    if (initialize_mqtt(&client) != 0)
    {
        return EXIT_FAILURE;
    }

    // 主循环
    while (1)
    {
        handle_reporting_mode(client, pubtype, period);
    }

    // 清理资源（正常情况下不会执行到这里）
    MQTTClient_unsubscribe(client, TOPIC_DOWN);
    MQTTClient_disconnect(client, TIMEOUT);
    MQTTClient_destroy(&client);

    if (Pointroot)
        cJSON_Delete(Pointroot);
    shm_del(&para);

    return EXIT_SUCCESS;
}